package com.sub.spark.sql.udf;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.api.java.UDF2;
import org.apache.spark.sql.types.DataTypes;

/**
 * @ClassName SparkSqlUDF
 * @Description: UDF:用户定义函数，一行进入，一行出
 * @Author Submerge.
 * @Since 2025/5/24 13:28
 * @Version 1.0
 */
public class SparkSqlUDF {

    public static void main(String[] args) {

        SparkSession sparkSession = SparkSession
                .builder()
                .appName("sub-spark-sql-udf")
                .master("local[2]")
                .getOrCreate();


        Dataset<Row> json = sparkSession.read().json("data/demo/spark/user.json");


        json.createOrReplaceTempView("user");

        //自定义函数并注册：该函数接受两个参数，拼接返回一个字符串
        sparkSession.udf().register("sub_concat",
                new UDF2<String, String, String>() {
                    @Override
                    public String call(String param1, String param2) throws Exception {
                        return param1 + param2;
                    }
                }
                , DataTypes.StringType);


        //使用SQL内置函数
        Dataset<Row> result = sparkSession.sql("select concat(\"姓名：\",name) as newName from user");
        result.show();
        System.out.println("=============");
        //使用自定义函数
        Dataset<Row> result2 = sparkSession.sql("select sub_concat(\"姓名：\",name) as newName from user");
        result2.show();


    }
}
